package com.gamejelly.proxy.client;

import com.gamejelly.proxy.pojo.RpcCall;
import com.gamejelly.proxy.pojo.RpcResponse;
import com.gamejelly.proxy.util.Timer;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.Setter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.*;

/**
 * Created by haoyifen on 2017/6/19 12:58.
 */
public class ConnectionManager {

    private static ConnectionManager Instance;


    private static final Logger logger = LoggerFactory.getLogger(ConnectionManager.class);

    private Bootstrap bootstrap;

    @Setter
    private EventLoopGroup eventLoopGroup;

    private final ConcurrentHashMap<String, Connection> conns = new ConcurrentHashMap<>();

    private final ConcurrentHashMap<String, CompletableFuture<RpcResponse>> futures = new ConcurrentHashMap<>();

    private final CountDownLatch latch = new CountDownLatch(1);

    public static ConnectionManager getInstance() {
        if (Instance == null) {
            Instance = new ConnectionManager();
            Instance.init();
        }
        return Instance;
    }

    public void oneWay(String address, RpcCall call) throws InterruptedException, TimeoutException {
        getConnection(address).oneway(call);
    }

    public CompletableFuture<RpcResponse> call(String address, RpcCall call) throws InterruptedException, TimeoutException {
        return getConnection(address).call(call);
    }

    private Connection getConnection(String address) throws InterruptedException {
        Connection connection = conns.get(address);
        if (connection == null) {
            Channel channel = createConn(address);
            connection = new Connection(channel);
            conns.put(address, connection);
            if (!connection.isActive()) {
                retry(connection, address);
            }
        }
        return connection;
    }

    private synchronized Channel createConn(String address) throws InterruptedException {
        String[] addr = address.split(":");
        ChannelFuture future = bootstrap.connect(addr[0], Integer.parseInt(addr[1])).sync();
        return future.channel();
    }


    private synchronized void init() {

        if (eventLoopGroup == null) {
            eventLoopGroup = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors());
        }
        bootstrap = new Bootstrap()
                .group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .handler(new ClientChannelInitializer())
                .option(ChannelOption.SO_KEEPALIVE, true);
        latch.countDown();
    }

    public void reConnected(Connection connection) {
        for (Map.Entry<String, Connection> entry : conns.entrySet()) {
            if (entry.getValue().equals(connection)) {
                String address = entry.getKey();
                retry(connection, address);
            }
        }
    }

    private void retry(Connection connection, String address) {
        boolean ok = false;
        try {
            Channel newChannel = createConn(address);
            connection.reConnected(newChannel);
            if (connection.isActive())
                ok = true;
        } catch (Exception ignore) {
        }
        if (!ok)
            Timer.INSTANCE.newTimeout(t -> {
                logger.info("try to connect : {}", address);
                retry(connection, address);
            }, 10, TimeUnit.SECONDS);
    }
}
